package cn.itcast.streaming.source;

import cn.itcast.streaming.bean.ElectricFenceModel;
import cn.itcast.streaming.bean.ElectricFenceRuleTmp;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.JDBCUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

public class MysqlElectricFenceResultSource extends RichSourceFunction<HashMap<String,Long>> {
    private static Logger logger = LoggerFactory.getLogger(MysqlElectricFenceResultSource.class);
    private ParameterTool parameterTool; //全局参数对象
    private Connection conn;
    private Statement statement;
    private Boolean isRunning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        conn = JDBCUtil.getConnection();
        statement = conn.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(statement!=null && !statement.isClosed()) statement.close();
        if(conn!=null && !conn.isClosed()) conn.close();
    }

    @Override
    public void run(SourceContext<HashMap<String, Long>> sourceContext) throws Exception {
        while (isRunning){
            String sql = "select vin,min(id) as id\n" +
                    "from vehicle_networking.electric_fence\n" +
                    "where inTime is not null and outTime is null\n" +
                    "group by vin;";
            ResultSet resultSet = statement.executeQuery(sql);
            HashMap<String, Long> map = new HashMap<>();
            while (resultSet.next()){
                map.put(resultSet.getString("vin"),resultSet.getLong("id"));
            }
            if(map.isEmpty()){
                logger.info("没有获取到电子围栏规则数据....");
            }else {
                logger.info("获取的规则数据为："+map);
            }
            resultSet.close();
            sourceContext.collect(map);
            //休眠时间一定要小于窗口滚动时间，否则拿到的是历史数据进行拉宽
            TimeUnit.SECONDS.sleep(10);
        }
    }


    @Override//取消数据读取
    public void cancel() {
        isRunning = false;
    }
}
